package com.showdor.springboot;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;

import java.util.Optional;

/**
 * @Author: lijb6
 * @Version 1.0
 */
//@Component
public class PartitionConsumer {
    private final Logger logger = LoggerFactory.getLogger(PartitionConsumer.class);
    ConsumerRecordDAO recorder = new SimpleRecordDAOProxy();

    //分区消费
    @KafkaListener(containerGroup = "group_test", topicPartitions = {@TopicPartition(topic = "netcross", partitions = {"0"})})
    public void onMessage(ConsumerRecord<?, ?> consumerRecord, Acknowledgment ack) {
        Optional<?> optional = Optional.ofNullable(consumerRecord.value());
        if (optional.isPresent()) {
            Object msg = optional.get();
            recorder.create(consumerRecord.key(), msg, ack);
            logger.info("partition=0,message:[{}]", msg);
        }
    }

    @KafkaListener(containerGroup = "group_test", topicPartitions = {@TopicPartition(topic = "netcross", partitions = {"1"})})
    public void onMessage1(ConsumerRecord<?, ?> consumerRecord, Acknowledgment ack) {
        Optional<?> optional = Optional.ofNullable(consumerRecord.value());
        if (optional.isPresent()) {
            Object msg = optional.get();
            recorder.create(consumerRecord.key(), msg, ack);
            logger.info("partition=1,message:[{}]", msg);
        }
    }
}
